package com.navinfo.tripanalysis.offline.service.impl;

import com.google.common.collect.Lists;
import com.navinfo.tripanalysis.common.pojo.PointProtocol;
import com.navinfo.tripanalysis.common.util.DateUtils;
import com.navinfo.tripanalysis.offline.pojo.DataCleaningInfo;
import com.navinfo.tripanalysis.offline.service.PointProtocolConvertService;
import com.navinfo.tripanalysis.offline.service.SavePointDataCleaningService;
import lombok.Data;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.StructType;
import scala.Tuple2;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 数据清洗：落盘公共轨迹协议
 */
@Data
public class SavePointDataCleaningServiceImpl extends AbstractHiveSaveService implements SavePointDataCleaningService {

    /**
     * 适配统一协议落盘数据结构服务
     */
    private PointProtocolConvertService service;

    @Override
    public void save(SparkSession spark, JavaPairRDD<Long, List<PointProtocol>> pairTidOutData, long d) {
        long start = System.currentTimeMillis();
        logger.error("进行统一协议信息落盘，保存Hive开始...");

        String day = DateUtils.format(new Date(d), DateUtils.DateFormat.YYYYMMDD);

        //保存数据
        super.saveData(spark, pairTidOutData, null, true, service.getPartitionCols(),Lists.newArrayList(day,"tid%32"));

        logger.error("进行统一协议信息落盘，保存Hive数据结束，耗{}ms", System.currentTimeMillis()-start);
    }

    @Override
    protected StructType createSchema() {
        return service.createSchema(service.createStructTypeList());
    }

    @Override
    protected JavaRDD<Row> createRowRDD(JavaPairRDD resourceRDD) {
        return resourceRDD.filter((Function<Tuple2<Long,List<PointProtocol>>, Boolean>)  l -> !l._2.isEmpty())
                .flatMap((FlatMapFunction<Tuple2<Long,List<PointProtocol>>,Row>)  f -> {
                    List<PointProtocol> data = f._2;
                    return Optional.ofNullable(data).orElse(new ArrayList<>())
                            .stream().map(e -> service.toRow(e)).collect(Collectors.toList()).iterator();
                });
    }
}
